package com.gis507.mapreduce.intertest;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.Utils.UtilsClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.junit.Test;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class AISInterpolationReducer extends Reducer<Text, AisClassifyBean,AisClassifyBean, NullWritable> {

//    //设定的最大时间
//    private static final double STANDTIME = 30;
//    private static final double MinTime = 30;
//    //设定的最大距离
//    private static final double STANDSTANCE = 1000;
//    private static int sum = 0;
    private double maxStance = 500;

    @Override
    protected void reduce(Text key, Iterable<AisClassifyBean> values, Context context) throws IOException, InterruptedException {



            //实例化一个列表，用于存放数据
            List<AisClassifyBean> aisList = new ArrayList<>(1000);

            //遍历values
            for (AisClassifyBean value : values) {

                //将value添加到列表
                aisList.add(UtilsClass.getNewAisClassifyBean(value));

            }

            log.info("aisList的长度是：" + aisList.size());

            //所有插值得到的数据放到一个列表中
            ArrayList<AisClassifyBean> allAisInterList = new ArrayList<>();

            //所有满足的数据
            ArrayList<AisClassifyBean> rightAisList = new ArrayList<>();

//            //aisList是一个放有数据的列表
//            for (int i = 0; i < aisList.size() - 1; i++) {
//
//                //判断两点之间的差是否大于规定的最大时间差
//                //java中取绝对值 Math.abs()
//                if (Math.abs(aisList.get(i + 1).getUnixTime() - aisList.get(i).getUnixTime()) < STANDTIME) {
//
//                    //判断是否距离超限，如果超限就插值，不超限就说明可能是停靠，就可以分段
//                    if (UtilsClass.getStance(aisList.get(i), aisList.get(i + 1)) > STANDSTANCE) {
//
//                        //往数据里插值
//                        ArrayList<AisClassifyBean> aisInterList = UtilsClass.insertPoint(aisList.get(i), aisList.get(i + 1),STANDSTANCE);
//                        //将本次插值的数据放到总的插值列表中
//                        allAisInterList.addAll(aisInterList);
//
//                        //sum
////                    sum += 1;
//                    } else {
//                        //小于距离阈值，删除中间的点，直至输出的都是距离平均的点
//                        for (int j = i+1;j < aisList.size() - 1;j++){
//                            if (UtilsClass.getStance(aisList.get(i), aisList.get(j)) > STANDSTANCE){
//                                rightAisList.add(UtilsClass.getNewAisClassifyBean(aisList.get(j)));
//                                i = j;
//                                break;
//                            }
//                        }
//                    }
//
//                } else {
//                    //满足的数据列表
//                    rightAisList.add(UtilsClass.getNewAisClassifyBean(aisList.get(i)));
//                    log.info("符合条件的添加成功一条数据");
//                }
//            }

            for (int i = 0;i < aisList.size() - 2;i++){
                if (UtilsClass.getStance(aisList.get(i),aisList.get(i+1)) < 10000){
                    if (UtilsClass.getStance(aisList.get(i),aisList.get(i+1)) < maxStance){

                        for (int j = i + 1;j < aisList.size() - 2;j++){

                            if (UtilsClass.getStance(aisList.get(i),aisList.get(j)) > maxStance){

                                i= j - 1;
                                log.info("去除了密集的点，成功选取出两个有用点！");
                                rightAisList.add(UtilsClass.getNewAisClassifyBean(aisList.get(j)));
                                break;
                            }
                        }
                    }else{

                        ArrayList<AisClassifyBean> aisInterList = UtilsClass.insertPoint(aisList.get(i), aisList.get(i + 1),maxStance);
                        //将本次插值的数据放到总的插值列表中
                        allAisInterList.addAll(aisInterList);
                    }
                }else{

                    log.info("这是一个异常点");
                }


            }

            //汇总数据
            rightAisList.addAll(allAisInterList);

            //写出
            for (AisClassifyBean aisInterBean : rightAisList) {

                context.write(aisInterBean,NullWritable.get());
            }


    }
}
